package org.example;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.util.List;

/**
 * @author zhangjg
 * @desc
 * @date 2023/03/30 20:29
 **/
public class CTest {

    private static final Logger logger = LoggerFactory.getLogger(CTest.class);

    public static void main(String[] args) throws Exception{
// Initialize Consumer and set Consumer Goup Name
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg1");

        // Set the address of NameServer
        consumer.setNamesrvAddr("192.168.10.1:9876");
        // Subscribe One or more of topics，and specify the tag filtering conditions, here specify * means receive all tag messages
        consumer.subscribe("publicTopic", "*");
        consumer.subscribe("userTopic", "*");
        // Register a callback interface to handle messages received from the Broker
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                msgs.forEach( msg -> {
//                    System.out.println(new String(msg.getBody()));
//                    logger.info("v1={},v2={},body={}",msg.getUserProperty("key"),msg.getUserProperty("x1"),new String(msg.getBody()));
                    logger.info(msg.toString());
                    logger.info("{},{}",msg.getUserProperty("simpUser"),new String(msg.getBody()));
                });
                // Return to the message consumption status, ConsumeConcurrentlyStatus.CONSUME_SUCCESS for successful consumption
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // Start Consumer
        consumer.start();
        System.out.printf("Consumer Started.%n");

    }
}
